#include "app_pool.h"
#include "mqueue.h"
#include "pthread.h"
#include "log/log.h"
#include "stdlib.h"

static mqd_t mq_fd;                // 消息队列文件描述符
static char *mq_name = "/mq_pool"; // 消息队列名称
static pthread_t *threads_pool; // 线程池
static int pool_size; // 线程数量

int *task_thread(void *arg)
{
    Task task;
    //不断接受消息队列的任务处理
    while (1)
    {
        int len = mq_receive(mq_fd, (char *)&task, sizeof(Task), NULL);
        if(len == sizeof(Task))
        {
            task.task_fun(task.arg);
        }
        
    }
    
    return NULL;
}
int app_pool_init(int thread_num)
{
    // 消息队列属性
    struct mq_attr attr;
    attr.mq_maxmsg = 10;            // 最大消息数量
    attr.mq_msgsize = sizeof(Task); //消息最大长度
    //创建打开消息队列
    mq_fd = mq_open(mq_name, O_RDWR | O_CREAT, 0644, &attr);
    if (mq_fd == -1)
    {
        log_error("创建打开消息队列 error");
        return -1;
    }
    //启动多个线程
    pool_size = thread_num;
    threads_pool = (pthread_t *)malloc(thread_num * sizeof(pthread_t));
    for (int i = 0; i < thread_num; i++)
    {
        pthread_create(&threads_pool[i], NULL, (void *)task_thread, NULL);
    }
    
    return 0;
}

void app_pool_destroy(void)
{
    //销毁消息队列
    mq_close(mq_fd);
    mq_unlink(mq_name);
    //取消停止所有线程
    for (int i = 0; i < pool_size; i++)
    {
        pthread_cancel(threads_pool[i]);
        pthread_join(threads_pool[i], NULL);
    }
    //释放线程池内存
    free(threads_pool);
}

int app_pool_registerTask(int (*task_fun)(void *arg), void *arg)
{
    //创建任务
    Task task = {
        .arg = arg,
        .task_fun = task_fun,
    };
    //发送任务给消息队列
    return mq_send(mq_fd, (char *)&task, sizeof(Task), 1);
}
